use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};

use pdata::internal::data::protogen::collector::trace::v1::ExportTraceServiceRequest;
use pdata::internal::data::protogen::trace::v1::{ResourceSpans, ScopeSpans, Span};

use rust_collector::exporter::wal::{WalDataType, WalWriter};
use prost::Message;
use std::sync::Arc;
use tokio::sync::Mutex;

fn unique_temp_dir() -> PathBuf {
    let mut dir = std::env::temp_dir();
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    let pid = std::process::id();
    dir.push(format!("rust_collector_wal_test_{}_{}", pid, nanos));
    dir
}

fn minimal_trace_request() -> ExportTraceServiceRequest {
    let span = Span {
        trace_id: vec![0; 16],
        span_id: vec![0; 8],
        trace_state: String::new(),
        parent_span_id: Vec::new(),
        flags: 0,
        name: "test_span".to_string(),
        kind: 1, // INTERNAL
        start_time_unix_nano: 1,
        end_time_unix_nano: 2,
        attributes: Vec::new(),
        dropped_attributes_count: 0,
        events: Vec::new(),
        dropped_events_count: 0,
        links: Vec::new(),
        dropped_links_count: 0,
        status: None,
    };

    let scope_spans = ScopeSpans {
        scope: None,
        spans: vec![span],
        schema_url: String::new(),
    };

    let resource_spans = ResourceSpans {
        resource: None,
        scope_spans: vec![scope_spans],
        schema_url: String::new(),
    };

    ExportTraceServiceRequest {
        resource_spans: vec![resource_spans],
    }
}

#[tokio::test]
async fn wal_append_and_stream_read_traces() {
    let root = unique_temp_dir();
    tokio::fs::create_dir_all(&root).await.unwrap();
    let wal = WalWriter::new(root.clone());

    let req = minimal_trace_request();
    wal.append(&req).await;

    let frames: Arc<Mutex<Vec<Vec<u8>>>> = Arc::new(Mutex::new(Vec::new()));
    let frames_cloned = frames.clone();
    wal.read_frames_streaming(WalDataType::Traces, 0, move |payload| {
        let frames_cloned = frames_cloned.clone();
        async move {
            frames_cloned.lock().await.push(payload);
        }
    })
    .await
    .unwrap();

    let frames_guard = frames.lock().await;
    assert_eq!(frames_guard.len(), 1, "should read exactly one frame");

    let decoded = ExportTraceServiceRequest::decode(frames_guard[0].as_slice()).unwrap();
    assert_eq!(decoded.resource_spans.len(), 1);
    assert_eq!(decoded.resource_spans[0].scope_spans.len(), 1);
    assert_eq!(decoded.resource_spans[0].scope_spans[0].spans.len(), 1);
}

#[tokio::test]
async fn wal_update_state_and_compact_keeps_unsent_frames() {
    let root = unique_temp_dir();
    tokio::fs::create_dir_all(&root).await.unwrap();
    let wal = WalWriter::new(root.clone());

    let req1 = minimal_trace_request();
    let req2 = minimal_trace_request();

    // append two frames
    wal.append(&req1).await;
    wal.append(&req2).await;

    // mark first frame as sent
    let first_frame_size = 4u64 + req1.encoded_len() as u64;
    let ok = wal.update_sent_state(WalDataType::Traces, first_frame_size).await;
    assert!(ok, "update_sent_state should succeed");

    // compact: remove sent bytes
    wal.compact(WalDataType::Traces).await.unwrap();

    // state should reset to 0 and exactly one frame remains
    let state = wal.read_state(WalDataType::Traces).await.unwrap();
    assert_eq!(state.sent_index, 0);

    let frames: Arc<Mutex<Vec<Vec<u8>>>> = Arc::new(Mutex::new(Vec::new()));
    let frames_cloned = frames.clone();
    wal.read_frames_streaming(WalDataType::Traces, 0, move |payload| {
        let frames_cloned = frames_cloned.clone();
        async move {
            frames_cloned.lock().await.push(payload);
        }
    })
    .await
    .unwrap();

    let frames_guard = frames.lock().await;
    assert_eq!(frames_guard.len(), 1, "one unsent frame should remain after compaction");
    let decoded = ExportTraceServiceRequest::decode(frames_guard[0].as_slice()).unwrap();
    assert_eq!(decoded.resource_spans.len(), 1);
}

#[tokio::test]
async fn wal_paths_helpers_work() {
    let root = unique_temp_dir();
    let traces = WalDataType::Traces.wal_path(&root);
    let tstate = WalDataType::Traces.state_path(&root);
    assert!(traces.ends_with("traces.wal"));
    assert!(tstate.ends_with("traces.state"));
}


